/*
 * Copyright 2012 Google Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.google.dspl.validate.validator;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.dspl.model.Concept;
import com.google.dspl.model.Concept.ConceptTableMapping;
import com.google.dspl.model.Dataset;
import com.google.dspl.model.Identifier;
import com.google.dspl.model.Slice;
import com.google.dspl.model.Slice.TableMapping;
import com.google.dspl.model.Table.Column;
import com.google.dspl.validate.ItemReference;
import com.google.dspl.validate.ValidationContext;
import com.google.dspl.validate.violation.TableDefinesConceptAndUsedInSliceViolation;
import com.google.dspl.validate.violation.TableDefinesMultipleConceptsViolation;
import com.google.dspl.validate.violation.TableUsedByMultipleSlicesViolation;
import com.google.dspl.validate.violation.Violation;

import java.util.Collection;
import java.util.List;
import java.util.Set;

/**
 * {@link Validator} for the {@link Concept} mappings in a {@link Dataset}, in particular, looking
 * for violations in mappings from {@link Column}s to Concepts when considering all of these
 * mappings across an entire {@link Dataset}.
 *
 * @author James Pack
 */
public class DatasetConceptMappingsValidator implements Validator<Dataset> {

  // Encapsulation of the state of a single validate call.  Validators are immutable; all of their
  // state is on the call stack.  This class encapsulates the state to make it easier to manage.
  private static final class State {
    private final Multimap<Identifier, Concept> conceptDefinitionTables = HashMultimap.create();
    private final Multimap<Identifier, Slice> sliceTables = HashMultimap.create();
  }

  @Override
  public List<Violation> validate(Dataset dataset, ValidationContext context) {
    State state = new State();
    findConceptDefinitionTables(state, dataset);
    findSliceTables(state, dataset);

    ImmutableList.Builder<Violation> violations = ImmutableList.builder();
    violations.addAll(validateMixedUsage(state, dataset, context));
    violations.addAll(validateConceptDefinitions(state, dataset, context));
    violations.addAll(validateSliceDefinitions(state, dataset, context));

    return violations.build();
  }

  /**
   * Check if any table is being used to define more than one concept.
   */
  private List<Violation> validateConceptDefinitions(State state, Dataset dataset,
      ValidationContext context) {
    List<Violation> violations = Lists.newArrayList();
    for (Identifier tableId : state.conceptDefinitionTables.keys()) {
      Collection<Concept> concepts = state.conceptDefinitionTables.get(tableId);
      if (concepts.size() > 1) {
        violations.add(new TableDefinesMultipleConceptsViolation(
            tableReference(dataset, context, tableId), tableId.getObjectId()));
      }
    }
    return violations;
  }

  /**
   * Check if any table is being used to define more than one slice.
   */
  private List<Violation> validateSliceDefinitions(State state, Dataset dataset,
      ValidationContext context) {
    List<Violation> violations = Lists.newArrayList();
    for (Identifier tableId : state.sliceTables.keys()) {
      Collection<Slice> slices = state.sliceTables.get(tableId);
      if (slices.size() > 1) {
        violations.add(new TableUsedByMultipleSlicesViolation(
            tableReference(dataset, context, tableId), tableId.getObjectId()));
      }
    }
    return violations;
  }

  /**
   * Check if a single table that are marked as defining slices and concepts.
   */
  private List<Violation> validateMixedUsage(State state, Dataset dataset,
      ValidationContext context) {
    List<Violation> violations = Lists.newArrayList();
    Set<Identifier> overlap = Sets.intersection(state.conceptDefinitionTables.keySet(),
        state.sliceTables.keySet());
    if (!overlap.isEmpty()) {
      for (Identifier tableId : overlap) {
        violations.add(new TableDefinesConceptAndUsedInSliceViolation(
            tableReference(dataset, context, tableId), tableId.getObjectId()));
      }
    }
    return violations;
  }

  private void findConceptDefinitionTables(State state, Dataset dataset) {
    for (Concept concept : dataset.getConcepts()) {
      ConceptTableMapping tableMapping = concept.getTable();
      if (tableMapping != null) {
        String tableId = tableMapping.getTableId();
        if (tableId != null) {
          Identifier identifier = new Identifier(dataset.getDatasetId(), tableId);
          state.conceptDefinitionTables.put(identifier, concept);
        }
      }
    }
  }

  private void findSliceTables(State state, Dataset dataset) {
    for (Slice slice : dataset.getSlices()) {
      TableMapping tableMapping = slice.getTable();
      if (tableMapping != null) {
        String tableId = tableMapping.getTableId();
        if (tableId != null) {
          Identifier identifier = new Identifier(dataset.getDatasetId(), tableId);
          state.sliceTables.put(identifier, slice);
        }
      }
    }
  }

  private ItemReference tableReference(
      Dataset dataset, ValidationContext context, Identifier tableId) {
    return ItemReference.reference(dataset, context.getDatasetResolver().getTable(tableId));
  }
}
